/*
 * File: ShardAlgorithm.java Author: Fating <xingon@163.com>
 * 
 * Created on 2014年10月29日上午9:26:26
 * 
 * Copyright (c) 2014, www.vip.com All Rights Reserved.
 */
package com.tc.vms.kafka;

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

import java.util.Random;

public class KafkaShardAlgorithm implements Partitioner {

    public KafkaShardAlgorithm(VerifiableProperties properties) {

    }

    @Override
    public int partition(Object key, int partionCount) {
        int shard = 0;
        String keyText = (String) key;
        if (keyText == null || keyText.trim().length() == 0) {
            shard = new Random().nextInt(partionCount);
            return shard;
        }
        char begin = keyText.charAt(0);
        if (begin > 57 || begin < 48) {
            shard = Math.abs(getHash(keyText) % partionCount);
            return shard;
        } else {
            try {
                shard = Integer.parseInt((String) key);
                shard = shard > -1 ? shard : 0;
                return shard % partionCount;
            } catch (Exception e) {
                shard = Math.abs(getHash(keyText) % partionCount);
                return shard;
            }
        }
    }

    private int getHash(String s) {
        return s.hashCode();
    }
}
